Skip to main content

RSMS results storage — implementation plan (Parquet / DuckDB era)

Archived. Historical Parquet + DuckDB validation phased plan — do not implement. Canonical today: rsms-worker-function/rsms_results_storage/ orchestrated by write_scenario_results_to_azure; see rsms-results-storage-design.

Status (historical archive): The Parquet triple (cxplt_time, cxplt_distance, mass_balance.parquet) and DuckDB validation flows in §4 Phases B–D below are retired for the product path. Normative storage design: rsms-results-storage-design. Migration / sequencing: rsms-results-table-storage-rework-plan (Phase 1 writer replaces dual Parquet with Azure Table float32 rows + minified JSON blobs).

This document remains useful for package layout (rsms-worker-function/rsms_results_storage/), parser reuse from file_parsers.py, and historical phase structure — adjust exit criteria to match table writers instead of write_results_parquet_set. The RSMS worker (rsms-worker-function, queue-driven or CLI) remains the integration point for both engine execution and results persistence in production (rsms-results-table-storage-rework-plan Phase 1b).

This plan originally turned the Parquet-era storage design into code. The canonical package lives under rsms-worker-function/rsms_results_storage/ (owned by the worker repo — no separate top-level library project; reuse is only within the worker unless needs change). It assumed Python 3.10+, PyArrow for Parquet writes, and optional DuckDB for validation queries.

Companion doc: rsms-results-storage-design

1. Recommended work surface: modules + thin entrypoints (not “notebook-first”)​

OptionProsCons
Notebook onlyFast explorationHard to import in the production worker layout; copy-paste drift
Standalone .py onlyRunnable from CLILess convenient for ad-hoc plots
Shared package + script + notebookOne implementation; notebook imports the same functionsSlightly more structure up front

Decision: Implement core logic in importable Python modules under a single package path (see §3). Add:

  • A thin CLI (e.g. python -m ... write-parquet --results-dir ...) for batch runs, or
  • A minimal notebook that only imports the package, calls public functions, and plots — no business logic in notebook cells.

That matches “you do the heavy lifting in code” while letting you use a notebook for visualization. The worker imports from rsms_results_storage (e.g. from rsms_results_storage.pipeline import ... — align PYTHONPATH / package layout under rsms-worker-function).

2. Design principles (worker-ready)​

  1. Pure functions where possible: (parsed_tables | paths) → pyarrow.Table or → None (upload side effects isolated).
  2. No global mutable config: pass RunConfig-like or explicit ResultsStorageConfig dataclass (paths, run_base, riverbasin_id, scenario_id, tolerances for future CTPLT).
  3. Stable public API: a small set of entrypoints:
    • load_cxplt_from_plt(path, timesteps) -> CXPLTFrame (name TBD)
    • cxplt_to_sparse_tables(...) -> tuple[pa.Table, pa.Table] (time-first, distance-first)
    • load_mass_balance_from_mas(...) -> pa.Table
    • write_results_parquet_set(out_dir, tables, ...) (local dir first; upload Parquet separately)
  4. Reuse existing parsers: refactor or wrap rsms-worker-function/file_parsers.py (parse_tim, parse_plt, parse_mas) so they can return columnar structures or feed the new layer — avoid duplicating .PLT format logic.
  5. I/O boundaries: Inputs are always local paths (engine output directory on the worker host, or a dev folder). Azure Blob is outputs only: upload Parquet to rsms-results, and optionally upload raw engine files to engine-files — use a BlobSink-style helper (upload_file(local, key) / upload_bytes). No blob download for aggregation inputs in the worker; the engine has already written the files locally.
  6. Schema: column names and dtypes match ssot/rsms-ssot.json once it exists (§5); until then, constants in one module (SCHEMA_CXPLT = ...).

3. Repository layout (decided)​

Canonical: the implementation lives under rsms-worker-function/rsms_results_storage/ (subpackage; e.g. pipeline.py, azure_table_plume.py, …). The worker adds one import path and there is no separate top-level rsms-results-storage/ project in this monorepo — we do not expect to consume this code from other repositories.

rsms-suite/
  rsms-worker-function/
    rsms_results_storage/         # canonical package (Table + blob writers today)
      __init__.py
      config.py
      pipeline.py
      ...
    file_parsers.py               # existing; shared PLT/MAS helpers as needed
    ...

Dev / notebooks: rsms-notebooks/storage/ may keep a scratch copy or symlink for CLI/notebook runs, or point PYTHONPATH at the worker folder. Prefer one source of truth under rsms-worker-function/rsms_results_storage.

Notebook (rsms-notebooks/storage/results_parquet_dev.ipynb): imports only the package; no chart business logic in cells.

4. Implementation phases​

Phase A — SSOT stub and types (0.5–1 day)​

  • Add ssot/rsms-ssot.json (minimal: entity names + column lists + dtypes) or config.py enums mirroring the design doc until SSOT is formalized.
  • Define CXPLTRow / MassBalanceRow (dataclass or TypedDict) and hours, miles, concentration naming consistent with DuckDB examples in the design doc.

Exit: Single import path for column name strings; no magic strings scattered.

Phase B — CXPLT: parse → long table → sparse → dual Parquet (core)​

  1. Input: .TIM + .PLT (same semantics as file_parsers.parse_tim / parse_plt).
  2. Flatten mile_conc_by_time to a long list of (hours, miles, concentration) (floats).
  3. Sparse rule: drop rows with concentration == 0 (or abs(conc) < epsilon if floats demand — document epsilon in config).
  4. Sort copy A: (hours, miles) → cxplt_time.parquet.
  5. Sort copy B: (miles, hours) → cxplt_distance.parquet.
  6. PyArrow: pa.Table.from_pandas or column builders; Snappy or Zstd compression; row group size tuned experimentally.

Exit: Two files written locally; row counts << dense grid if PLT is sparse.

Phase C — Mass balance Parquet​

  1. Reuse parse_mas output or parse .MAS in the same package.
  2. Build mass_balance.parquet sorted by hours; decide dense vs drop-zero for PMASS/DZMASS (open item in design doc §12).

Exit: Third file; DuckDB SELECT COUNT(*), MIN(hours), MAX(hours) sanity check.

Phase D — Validation (DuckDB)​

  • validate.py: run the design doc §10 queries against the three files (local paths).
  • Optional: read_parquet + WHERE hours = ? on cxplt_time and compare row counts to a reference hour from raw PLT.

Exit: Scripted check you can run in CI later.

Phase E — Blob upload (outputs only)​

  • Upload Parquet to rsms-results and (if required by product) raw engine artifacts to engine-files, each under {riverbasin_id}/{scenario_id}/.
  • Inputs for parsing remain the local working directory after the engine run — not re-fetched from engine-files for this pipeline.
  • Idempotency: overwrite same blob names on retry (per design).
  • Use AZURE_STORAGE_CONNECTION_STRING (same pattern as existing results_blob_upload.py).

Exit: After a successful local run folder → Parquet blobs in rsms-results (and optional engine-files copies).

Phase F — Worker integration​

  1. Import from rsms_results_storage under rsms-worker-function in the engine/results pipeline after parse (see run_rsms_handlers / engine_workflow).
  2. Feature flag (env var): WRITE_RESULTS_PARQUET=1 to avoid breaking existing JSON-only responses until tested.
  3. Logging: blob URLs + row counts in pipeline result JSON.

Exit: A successful worker run produces Parquet in storage on success (historical Phase F target; product path today is Table + JSON, not Parquet).

Phase G — Tests​

  • Unit: sparse filter (input with zeros → output without).
  • Unit: dual sort orders produce identical multiset of non-zero rows.
  • Golden file (optional): small .PLT / .MAS fixtures under rsms-worker-function/tests/fixtures/ (or next to rsms_results_storage/).

5. ON HOLD — CTPLT (reference for later)​

Not implemented until product defines the interpolation algorithm. Keep this section as a checklist when unblocked. Persistent TODO (Sudhir code adaptation, one-time generation + storage like mass balance, edge logic vs concentration_tolerance): rsms-results-backlog §1.

ItemReference
Legacy algorithmrsms-legacy/BusinessLogic/CTPLTGenerator.cs — requires CXPLT full set ordered by time, distance; mile step rmStepSize; MinLimit cutoff
Outputsctplt_time.parquet, ctplt_distance.parquet (same dual-layout idea as CXPLT)
DownstreamLegacy edge/peak charts use CTPLTResults + Runs.SimParamsConcentrationtolerance — see design §2.3
Code shapectplt.py: interpolate_cxplt_to_ctplt(cxplt_long_df, config) -> pa.Table then reuse parquet_write.dual_sort
SSOTAdd ctplt_time, ctplt_distance entities

Dependency: CXPLT Parquet (or in-memory CXPLT long table) must be complete enough for interpolation — if CXPLT is sparse, interpolation must define behavior at missing (hour, mile) (treat as 0).

6. Risks and mitigations​

RiskMitigation
Sparse CXPLT breaks assumptions in a naive port of CTPLTGeneratorDocument fill-zero before interpolation; add Phase G tests when CTPLT un-holds
Duplicate parser logic vs file_parsers.pyExtract shared plt_parser module used by worker and storage package
Large Parquet filesRow groups + compression; monitor Elk-scale runs (design §12)

7. Open decisions (pre-coding)​

  1. Float epsilon for “zero” concentration when dropping rows.
  2. Mass balance: keep all hours or omit zero rows (design §12 item 4).

8. Document index​

DocRole
rsms-results-storage-designWhat to store (Table plume grid + JSON blobs, 0.2 mi step)
rsms-results-table-storage-rework-planMigration phases (writer + API)
This fileHow to implement package layout + historical Parquet phases (legacy)

Implementation plan — package home: rsms-worker-function/rsms_results_storage/.

Last updated on by jacobgqc